package com.atguigu.chapter09;

import com.atguigu.chapter5.source.WaterSensor;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternSelectFunction;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.SimpleCondition;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.time.Duration;
import java.util.List;
import java.util.Map;

/**
 * @ClassName: Flink01_CEP_demo
 * @Description:
 * @Author: kele
 * @Date: 2021/4/10 14:54
 *
 * CEP编程
 *
 *  总体流程：
 *      1、定义匹配方式：Pattern.<泛型>begin("xxx")
 *      2、将cep应用在流上， CEP.pattern(流, 匹配方式);
 *      3、获取满足模式的数据： .select(new PatternSelectFunction<in, out>()
 *
 *
 * 单个模式
 *      --固定次数  .times(n)
 *      --范围内的次数   .times(x,x+y)
 *      --一次或多次     .oneormore
 *       --多次以上       .timeormore(n)
 *
 * 条件
 *      --迭代条件          IterativeCondition
 *      --简单条件          SimpleCondition
 *      --组合条件          .or(new ...)
 *      --中止条件          .until(new ....)
 * 组合模式
 *      --严格连续          .next()
 *      --松散连续          .followedBy
 *      --非确定松散连续     .followedByAny
 *   循环模式的连续性
 *      --严格连续        .consecutive
 *      --松散连续        .times(n)
 *      --非确定松散连续  .allowCombinations()
 * 贪婪性
 *      .greedy
 *
 * 模式的可选性
 *      .optional()  //可以不包含该模式
 *      .times(2).optional()   //匹配 2次或 零次
 * 模式组
 *           .begin(Pattern
 *                .<WaterSensor>begin("xxxx")
 *                .where(new SimpleCondition<WaterSensor>() {
 *                    @Override
 *                    public boolean filter(WaterSensor value) throws Exception {
 *                        return xxxxxxxxxxx;
 *                    }
 *                })
 *                .next("xxxx")
 *                .where(new SimpleCondition<WaterSensor>() {
 *                    @Override
 *                    public boolean filter(WaterSensor value) throws Exception {
 *                        return xxxxxxxxxxx;
 *                    }
 *                }))
 *
 * 超时数据
 *
 *      .within(Time.seconds(2));
 *
 * 匹配后跳过策略
 *
 *  	NO_SKIP: 每个成功的匹配都会被输出。
 *  	SKIP_TO_NEXT: 丢弃以相同事件开始的所有部分匹配。
 *  	SKIP_PAST_LAST_EVENT: 丢弃起始在这个匹配的开始和结束之间的所有部分匹配。
 *  	SKIP_TO_FIRST: 丢弃起始在这个匹配的开始和第一个出现的名称为PatternName事件之间的所有部分匹配。
 *  	SKIP_TO_LAST: 丢弃起始在这个匹配的开始和最后一个出现的名称为PatternName事件之间的所有部分匹配。
 *
 **/
public class Flink01_CEP_demo {

    public static void main(String[] args) {

        Configuration conf = new Configuration();
        conf.setInteger("rest.port",20000);

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);

        env.setParallelism(1);

        SingleOutputStreamOperator<WaterSensor> ds = env.readTextFile("in/sensor.txt")
                .map(new MapFunction<String, WaterSensor>() {

                    @Override
                    public WaterSensor map(String value) throws Exception {
                        String[] split = value.split(",");
                        return new WaterSensor(split[0],
                                Long.parseLong(split[1]) * 1000,
                                Integer.parseInt(split[2]));
                    }
                })
                .assignTimestampsAndWatermarks(WatermarkStrategy
                        .<WaterSensor>forBoundedOutOfOrderness(Duration.ofSeconds(5))
                        .withTimestampAssigner((element, recordTimestamp) -> element.getTs()));

        /**
         * 匹配方式
         */
        Pattern<WaterSensor, WaterSensor> pattern = Pattern
                .<WaterSensor>begin("start")
                .where(new SimpleCondition<WaterSensor>() {
                    @Override
                    public boolean filter(WaterSensor value) throws Exception {
                        return "sensor_1".equalsIgnoreCase(value.getId());
                    }
                }).times(2).consecutive();

        CEP.pattern(ds,pattern)
                .select(new PatternSelectFunction<WaterSensor, String>() {
                    @Override
                    public String select(Map<String, List<WaterSensor>> map) throws Exception {
                        return map.toString();
                    }
                })
                .print();


        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }


    }
}
